package com.kafka.consumer.mainstay;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author xuhaifeng
 * @since 2022年02月18日 下午 10:38
 */
@Slf4j
@Component
public class ConsumerRoute {
    private final StringRedisTemplate redis;

    @Autowired
    public ConsumerRoute(StringRedisTemplate redis) {this.redis = redis;}

    @KafkaListener(topics = {"lattice"})
    public void listener(ConsumerRecord<String, Object> record) {
        Optional.of(record.key()).ifPresent(key -> {
            if (Boolean.FALSE.equals(redis.hasKey(key))) {
                Optional.ofNullable(record.value()).ifPresent(o -> log.info(String.valueOf(o)));
                redis.opsForValue().set(key, "123");
            } else {
                log.info("信息已消费！！");
            }
        });
    }

}
